package com.novel.CanalClient;

import com.alibaba.fastjson.JSONArray;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.util.StringUtils;
import com.alibaba.fastjson.JSONObject;

@Slf4j
//@Component
public class CanalKafkaClient {
    @Autowired
    StringRedisTemplate redisTemplate;

    @KafkaListener(topics = "canal_topic")
    public void receive(ConsumerRecord<?, ?> consumer) {
        pross(consumer.value().toString());
    }

    private void pross(String json) {
        log.info(json);
        if (StringUtils.isEmpty(json)) {
            return;
        }
        JSONObject jsonObject = JSONObject.parseObject(json);
        String type = jsonObject.getString("type");
        JSONArray datas = jsonObject.getJSONArray("data");
        String database = jsonObject.getString("database");
        String table = jsonObject.getString("table");
        for (int i = 0; i < datas.size(); i++) {
            String key = database + ":" + table + ":" + datas.getJSONObject(i).get("id");
            String value = datas.getJSONObject(i).toString();
            log.info(type);
            switch (type) {
                case "INSERT":
                case "UPDATE":
                    update(key, value);
                    break;
                case "DELETE":
                    delete(key);
                    break;
            }
        }
    }

    private void update(String key, String value) {
        redisTemplate.opsForValue().set(key, value);
    }

    private void delete(String key) {
        redisTemplate.delete(key);
    }
}

